Skip to content

[Platform] Introduce asynchronous jobs and move MiniMax polling out of the result converter - #2408

Merged
chr-hertel merged 1 commit into
symfony:mainfrom
wachterjohannes:feature/platform-async-jobs
Sep 20, 2026
Merged

chr-hertel merged 1 commit into
symfony:mainfrom
wachterjohannes:feature/platform-async-jobs

Conversation

@wachterjohannes

Copy link
Copy Markdown
Member
Q A
Bug fix? no
New feature? yes
Docs? yes
Issues -
License MIT

Asynchronous work already exists in this component — hand-rolled, and in the wrong layer:

  • Bridge\MiniMax\MiniMaxResultConverter::handleAsyncTask() polls a task with clock->sleep() up to 600 times, holding an HttpClientInterface, an API key and a ClockInterface to do so — inside a class whose contract is a pure RawResult → Result mapping.
  • Bridge\Replicate\Client::request() does the same in the model client.
  • Capability::TEXT_TO_SPEECH_ASYNC is already in the enum, used by exactly one bridge.

So the question is not whether the component should support async work, but that it already does, twice, inconsistently. Today one invoke() call blocks for minutes, the caller cannot observe the job or stop waiting, the operation does not survive the process that started it, and the state machine is not testable in isolation.

This PR introduces the job as a platform primitive and moves MiniMax onto it. Replicate and batch endpoints are deliberately left for follow-ups.

Why "job" and not "batch"

Video generation, asynchronous speech, OpenAI/Anthropic message batches, Gemini long-running operations and Replicate predictions all share one mechanic: submit → identifier → poll → fetch. And the body you eventually get back is the same one a synchronous call would have returned, which is why ResultConverterInterface is untouched here — an asynchronous result is not a new result type, only a different delivery.

Batch is then "a job with N items" rather than a second abstraction: JobHandle::$data carries the per-item keys when that lands.

What it looks like

$handle = $platform->invoke('MiniMax-Hailuo-02', new Text('A cat playing the piano'), [
    'duration' => 6,
])->asJob();

The handle holds no client and no connection, only what is needed to ask the provider about the job again — so it can be stored and picked up elsewhere:

// process one
$repository->save($id, json_encode($handle));

// a worker, possibly much later
$handle = JobHandle::fromArray(json_decode($repository->load($id), true));
$jobClient = $platform->getJobClient($handle);

if ($jobClient->getStatus($handle)->is(JobStateCase::SUCCEEDED)) {
    $jobClient->getResult($handle)->asFile('video.mp4');
}

To simply block, hand it to a runner — which owns the one polling loop left in the component and takes its budget from the caller, seconds for speech and minutes for video, instead of a constant inside a bridge:

$result = (new JobRunner(pollInterval: 1.0, maxPolls: 600))->wait($platform->getJobClient($handle), $handle);

Pieces

Class Role
Job\JobHandle Serializable reference: id, provider, provider-specific data. No client, no closure
Job\JobStateCase / Job\JobStatus Normalized state plus the provider's own wording, mirroring FinishReason/FinishReasonCase. An unknown state maps to UNKNOWN and counts as non-terminal, so a state a provider adds later does not abort a running job
Job\JobClientInterface getStatus() / getResult(), exactly one request per call, never sleeps
Job\JobRunner The blocking loop, with ClockInterface, poll interval and budget
Result\JobResult ResultInterface whose content is the handle; DeferredResult::asJob() reads it
Job\JobProviderInterface Optional capability, so ProviderInterface stays untouched

Provider takes an optional JobClientInterface as its last constructor argument, and stamps the provider name onto the handle after conversion — a converter cannot know the name its provider was registered under, since that is a Provider argument and Factory::createProvider() lets it be overridden.

Exception\JobFailedException carries the status; Exception\JobTimeoutException carries the handle, so a job that outlived its budget can be handed on rather than lost.

Verified against the live API

Not only against mocks — every path was run against api.minimax.io:

Result
Async speech submit 1.0s, finished after 8.5s, mp3 written
Video finished after 79.5s, 668 KB, ISO Media, MP4 Base Media v1
Resume started in one process, Processing in a second, Success + download in a third — resolved purely from the serialized handle

Running it also turned up two things mocks could not have shown, both fixed here:

MiniMax reports rejections with HTTP 200. The reason sits in base_resp and the payload keys are simply absent — an unknown voice on t2a_v2 gives 2054 "voice id not exist", an unsupported model on image_generation gives 2013 "invalid params, ...", an empty account gives 1008 "insufficient balance", all with HTTP 200, while every success carries status_code: 0. Previously this surfaced as an error about a missing array key — or, for an asynchronous task, as polling a task that never existed (task_id: 0 is handed out as a placeholder) until the budget ran out. The converter now checks base_resp for every endpoint. Chat is the exception: a bad model there does return a real HTTP 400, already covered by HttpStatusErrorHandlingTrait.

The asynchronous speech endpoint does not return audio. It returns a tar bundling the mp3 with a .titles and an .extra file, which the bridge previously labelled audio/mpeg — so asFile('out.mp3') wrote an archive. The job client now unpacks the audio, making async and sync speech produce the same thing. TarArchive is a small ustar reader: the archive uses the header's prefix field because the paths exceed the 100-byte name field, and the mp3 is not the first member, so both had to be handled. PharData would have meant a temp file plus ext-phar as a new bridge dependency.

The test fixture is a real archive produced by tar --format ustar rather than hand-written, so the reader is not being tested against its own writer.

Not in this PR

  • Replicate (Client.php:60) — same shape, follow-up now that the primitive exists.
  • Batch (N items, custom_id correlation).
  • No Capability::ASYNC_JOB — the JobResult type is already the signal, a capability would be a second truth.
  • No scheduler or webhook handling in the component. Polling belongs in ai-bundle via Messenger/Scheduler/symfony/webhook; Platform only provides submit/status/fetch.
  • No MiniMax status-code → exception mapping. Every non-zero base_resp code throws a RuntimeException carrying the code and the provider's message. Mapping 1004 to AuthenticationException and 1002 to RateLimitExceededException would be plausible, but only 1008, 2013 and 2054 were actually observed and a half-map is worse than none. Happy to add it if wanted.

BC break

Video generation and asynchronous speech return a JobResult instead of binary data, so reading them through asBinary()/asFile() throws. Waiting became explicit. MiniMaxResultConverter also lost its HTTP client, API key, endpoint and clock arguments. Both are documented in UPGRADE.md; code building the bridge through Bridge\MiniMax\Factory is unaffected.

@carsonbot carsonbot added Status: Needs Review Feature New feature Platform Issues & PRs about the AI Platform component labels Aug 15, 2026
@wachterjohannes
wachterjohannes force-pushed the feature/platform-async-jobs branch from c7ba9f6 to 3fae510 Compare August 15, 2026 20:25
*
* @internal
*/
final class TarArchive

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the API delivery a tar file - to avoid the dependency to a tar lib i have implemented this little class

Comment thread docs/components/platform.rst Outdated
Comment thread src/platform/src/Platform.php
@chr-hertel chr-hertel added the BC Break Breaking the Backwards Compatibility Promise label Aug 16, 2026
@chr-hertel chr-hertel added this to the v1.0 milestone Aug 23, 2026
@wachterjohannes
wachterjohannes force-pushed the feature/platform-async-jobs branch from 2ad2088 to 7d3bcd1 Compare August 24, 2026 21:21
@wachterjohannes
wachterjohannes force-pushed the feature/platform-async-jobs branch from 7d3bcd1 to 5bfd3b2 Compare September 1, 2026 19:44

@chr-hertel chr-hertel left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We have an isolation problem here that we should resolve still.

Symptoms:

  • $result instanceof JobResult in TypedResultTrait::as()
  • $result instanceof JobResult in Provider::invoke()

I think this should be handled already down below in the ModelClient/ResultConverter layer while constructing the JobResult. We don't have the provider name there, but strictly speaking that is irrelevant. from a logic point of view that layer "knows" where it is and how to move on 🤔

@chr-hertel chr-hertel left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's do this 💪

@chr-hertel
chr-hertel force-pushed the feature/platform-async-jobs branch from 9ac8e0e to 3bf8906 Compare September 20, 2026 22:27
@chr-hertel

Copy link
Copy Markdown
Member

Thank you @wachterjohannes.

@chr-hertel
chr-hertel merged commit 7f87acf into symfony:main Sep 20, 2026
36 checks passed
chr-hertel added a commit to chr-hertel/ai that referenced this pull request Sep 21, 2026
chr-hertel added a commit that referenced this pull request Sep 21, 2026
…r-hertel)

This PR was squashed before being merged into the main branch.

Discussion
----------

[Platform] Fix asynchronous job follow-ups from #2408

| Q             | A
| ------------- | ---
| Bug fix?      | no
| New feature?  | no
| Docs?         | yes
| Issues        | -
| License       | MIT

Follow-ups on #2408, found while reviewing it after the merge. Ships with #2408 in 0.14, so nothing
released changes and no `UPGRADE.md` entry is added — the existing ones are corrected to describe
what will actually ship.

* MiniMax answers a healthy poll with `status_code: 0` **and** `status_msg: "success"`, which was read as a failure message, so every running job carried `"success"` in `JobStatus::getError()`. Verified against the live API.
* A query MiniMax rejects with HTTP 200 carries the reason in `base_resp` and no state at all, which mapped to `UNKNOWN` (non-terminal) — the runner then polled it for the whole budget, up to ten minutes for a video handle. It now ends the job, while a state merely spelled in a way the bridge does not know stays non-terminal.
* `JobRunner` turned the budget into a poll count, so the requests themselves were free: `maxDuration: 600` meant 600 requests plus 599 sleeps. It now waits against a clock deadline.
* `JobClientInterface::supports()` had no caller anywhere; the runner asks before polling.
* `MiniMaxResultConverter` held a `MiniMaxJobClient` only to read a name off it, building one with `new EventSourceHttpClient()` and an empty API key when given none. It takes the name; `MiniMaxJobClient::createHandle()` and its provider argument go.
* Docs: the autowiring example used `$minimaxJobClient`, but the registered alias is `JobClientInterface $minimax`, so it never autowired — checked by compiling a container. Also a nullable `getProvider()` passed to `ContainerInterface::get()`, an unassigned `$id`, and examples using `createProvider()` where every other MiniMax example uses `createPlatform()`.
* The 0.14 rebase had replaced the `CHANGELOG` heading underline with `==` in two files; `examples/minimax/.gitignore` missed the artifacts the new examples write.

All nine MiniMax examples were run against the live API: async speech (9s, real mp3 out of the tar),
video generation (83s, 792 KB mp4) and the resume flow across four processes from the stored handle.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Commits
-------

e7af258 [Platform] Fix asynchronous job follow-ups from #2408
chr-hertel added a commit to chr-hertel/ai that referenced this pull request Sep 21, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/
BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is
a job that happens to carry many requests.

Submitting goes through `Platform::invoke()` with several inputs instead of one,
keyed by the identifier each result is reported back under, and returns a plain
`JobHandle`:

    $handle = $platform->invoke('gpt-4o-mini', [
        'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')),
        'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')),
    ], ['batch' => true])->asJob();

Resolving it is `JobClientInterface`, so storing the handle, polling from a
worker, `JobRunner`, the exceptions and the bundle wiring all come for free.

A finished batch converts into a `BatchResult` of `BatchItem`s, read from the
download as they are iterated. A successful item carries the ordinary result the
request would have produced synchronously, built by the bridge's own converter,
so tool calls and structured output survive the detour.
@wachterjohannes
wachterjohannes deleted the feature/platform-async-jobs branch September 21, 2026 03:58
chr-hertel added a commit to chr-hertel/ai that referenced this pull request Sep 22, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/
BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is
a job that happens to carry many requests.

Submitting goes through `Platform::invoke()` with several inputs instead of one,
keyed by the identifier each result is reported back under, and returns a plain
`JobHandle`:

    $handle = $platform->invoke('gpt-4o-mini', [
        'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')),
        'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')),
    ], ['batch' => true])->asJob();

Resolving it is `JobClientInterface`, so storing the handle, polling from a
worker, `JobRunner`, the exceptions and the bundle wiring all come for free.

A finished batch converts into a `BatchResult` of `BatchItem`s, read from the
download as they are iterated. A successful item carries the ordinary result the
request would have produced synchronously, built by the bridge's own converter,
so tool calls and structured output survive the detour.
chr-hertel added a commit that referenced this pull request Sep 22, 2026
… (chr-hertel)

This PR was merged into the main branch.

Discussion
----------

[Platform] Convert polling bridges to asynchronous jobs

| Q             | A
| ------------- | ---
| Bug fix?      | no
| New feature?  | yes
| Docs?         | yes
| Issues        | -
| License       | MIT

Follow-up to #2408: the Higgsfield, Venice and Replicate bridges solved async work with their own
polling loops. They now return a `JobResult` carrying a `JobHandle`, resolved through a per-bridge
`JobClientInterface` — same shape as MiniMax.

- `HiggsfieldJobClient`, `VeniceJobClient`, `ReplicateJobClient` + `Factory::createJobClient()`
- ai-bundle registers `ai.platform.job_client.higgsfield` / `.venice` (tagged + autowired)
- Removed the now-dead `clock` / `pollingInterval` / `max_polling_attempts` / `polling_interval_seconds` args

While at it, `JobHandle` gained `getPollInterval()` next to `getMaxDuration()` — how often a job is
worth asking about is the bridge's knowledge just like how long it runs — and `JobRunner` gained a
`maxPolls` ceiling, which is the caller's (a rate limit, an API bill). Both resolve
`call > runner > handle > default`.

**Needs the `BC Break` label** — Replicate is released, and its invocations no longer return a
`TextResult` directly (`UPGRADE.md` entry included). Higgsfield and Venice were both added in the
unreleased 0.14, so they get no upgrade entry.

Cassettes were extended by hand rather than re-recorded (no API keys): `getStatus()` + `getResult()`
means one extra poll of the finished job, so each gained a duplicate of its final terminal-status
interaction — the same shape MiniMax's committed cassette already has.

Also passes the examples' clock to the MiniMax job runners, which were sleeping for real on replay:
`minimax/text-to-video.php` went from 52s to 0.06s, and the whole replay suite from 67s to 16s.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Commits
-------

af06ee0 [Platform] Convert polling bridges to asynchronous jobs
chr-hertel added a commit to chr-hertel/ai that referenced this pull request Sep 22, 2026
An alternative to symfony#1802: batch is not its own stack of BatchJob/BatchManager/
BatchStatus classes, but the asynchronous job that symfony#2408 introduced - a batch is
a job that happens to carry many requests.

Submitting goes through `Platform::invoke()` with several inputs instead of one,
keyed by the identifier each result is reported back under, and returns a plain
`JobHandle`:

    $handle = $platform->invoke('gpt-4o-mini', [
        'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')),
        'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')),
    ], ['batch' => true])->asJob();

Resolving it is `JobClientInterface`, so storing the handle, polling from a
worker, `JobRunner`, the exceptions and the bundle wiring all come for free.

A finished batch converts into a `BatchResult` of `BatchItem`s, read from the
download as they are iterated. A successful item carries the ordinary result the
request would have produced synchronously, built by the bridge's own converter,
so tool calls and structured output survive the detour.
chr-hertel added a commit that referenced this pull request Sep 22, 2026
…nous jobs (chr-hertel)

This PR was squashed before being merged into the main branch.

Discussion
----------

[Platform] Add OpenAI batch requests on top of asynchronous jobs

| Q             | A
| ------------- | ---
| Bug fix?      | no
| New feature?  | yes
| Docs?         | yes
| Issues        | Fix #1776
| License       | MIT

An alternative to #1802: now that #2408 landed asynchronous jobs, a batch is just a job that carries many requests — no `BatchJob`/`BatchManager`/`BatchStatus` stack needed.

```php
$handle = $platform->invoke('gpt-4o-mini', [
    'capital-fr' => new MessageBag(Message::ofUser('What is the capital of France?')),
    'capital-de' => new MessageBag(Message::ofUser('What is the capital of Germany?')),
], ['batch' => true])->asJob();

// later, in a worker, from the stored handle
$jobClient = OpenAiFactory::createJobClient($apiKey);

foreach ($jobClient->getResult($handle)->getContent() as $item) {
    echo $item->getId().': '.($item->isSuccess() ? $item->getResult()->getContent() : $item->getError());
}
```

* Storing the handle, polling, `JobRunner`, the exceptions and the bundle wiring all come from #2408 unchanged
* `['batch' => true]` is handled by `Gpt\ModelClient` and delegated to `Batch\BatchClient`, like `MiniMaxClient` routes on `async` — no routing in `Provider`
* A finished batch converts into a `Result\BatchResult` of `Result\BatchItem`s, read from the download as they are iterated; a successful item carries the ordinary result the request would have produced synchronously, built by the bridge's own converter (tool calls, structured output, finish reasons included)
* An item without a result states *why* through a `Result\BatchItemCase` — `ERRORED` has to be fixed, `CANCELED`/`EXPIRED` were never sent and can simply be resubmitted — with the provider's own wording kept on `getRaw()`, the same split as `FinishReason` and `JobStatus`
* A cancelled or expired batch still hands out what it finished, since fetching keys on the result file rather than on `completed`
* `Batch\JobClient` adds `getRequestCounts()` and `cancel()` — on the bridge, not on `JobClientInterface`

### Does the result shape generalize?

Checked the two obvious next bridges before settling it, since `BatchResult`/`BatchItem` sit in the Platform:

* **Mistral** — the same envelope to the letter. Its SDK parses `custom_id` / `response.body` / `error` with `status_code >= 400` as the failure test, which is `Batch\JobClient::toItem()` line for line, and documents that envelope as invariant across endpoints. Statuses map onto `JobStateCase` without a gap.
* **Anthropic** — submits inline rather than as a file (submit side only, per bridge) and reports four per-request outcomes: `succeeded`, `errored`, `canceled`, `expired`. That is what `BatchItemCase` is for. Its batch-level status is only `in_progress`/`canceling`/`ended`, and a cancelled batch also ends as `ended` with partial results — so keying result-fetching on the file rather than on a success status is not just convenient here, it is the only thing that works there.

Request counts differ per provider (`{total, completed, failed}` vs `{processing, succeeded, errored, canceled, expired}` vs `{total, succeeded, failed}`), which is why `getRequestCounts()` stays on the bridge.

### Open point

No `Capability::BATCH`. OpenAI gates batching per endpoint, not per model, and the option is handled by the endpoint's own model client, so the gate is structural and there is no list to keep up to date — an ineligible model is rejected by the API. Happy to add the capability instead if you prefer the pre-flight check.

Scoped out for follow-ups: the Anthropic and Mistral bridges, and batched embeddings (same `BatchClient`, different endpoint).

Commits
-------

5e61785 [Platform] Add OpenAI batch requests on top of asynchronous jobs
marco-jouwweb pushed a commit to marco-jouwweb/symfony-ai that referenced this pull request Sep 23, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

BC Break Breaking the Backwards Compatibility Promise Feature New feature Platform Issues & PRs about the AI Platform component Status: Reviewed

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants